package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.article.config.kafka.KafkaStreamListener;
import com.heima.article.service.ApArticleService;
import com.heima.common.constans.message.MessageConstants;
import com.heima.model.article.pojos.ApArticle;
import com.heima.model.article.vo.HotArticleVo;
import com.heima.model.mess.app.UpdateArticleMess;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.springframework.beans.factory.annotation.Autowired;

public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String,String>> {
    @Override
    public String listenerTopic() {
        return MessageConstants.HOT_ARTICLE_BEHAVIOR_SEND_TOPIC;
    }

    @Override
    public String sendTopic() {

        return MessageConstants.HOT_ARTICLE_BEHAVIOR_RECEIVE_TOPIC;
    }

    @Autowired
    ApArticleService apArticleService;
    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {

        //阅读：1，点赞：3，评论：5，收藏：8
        KStream<String, String> map = stream.map((key,  value)-> {
                UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                ApArticle article = apArticleService.getById(mess.getArticleId());
                int i = 0;
                if (mess.getType().equals(UpdateArticleMess.UpdateArticleType.COLLECTION)) {
                    article.setCollection(article.getCollection() + mess.getAdd());
                } else if (mess.getType().equals(UpdateArticleMess.UpdateArticleType.COMMENT)) {
                    article.setComment(article.getComment() + mess.getAdd());
                } else if (mess.getType().equals(UpdateArticleMess.UpdateArticleType.LIKES)) {
                    article.setLikes(article.getLikes() + mess.getAdd());
                } else if (mess.getType().equals(UpdateArticleMess.UpdateArticleType.VIEWS)) {
                    article.setViews(article.getViews() + mess.getAdd());
                }

                int score = getScore(article);
                apArticleService.updateById(article);

                HotArticleVo vo = JSON.parseObject(JSON.toJSONString(article), HotArticleVo.class);
                vo.setScore(score);
                return new KeyValue<>(key, JSON.toJSONString(vo));
        });

      return map;
    }

    //计算分数
    //阅读：1，点赞：3，评论：5，收藏：8
    private int getScore(ApArticle article) {
        int i=article.getViews()+article.getComment()*3+article.getCollection()*8+article.getLikes()*3;
        return i*3;
    }

}
